package com.csw.flink.state

import java.util.Properties

import org.apache.flink.api.common.functions.{RichFlatMapFunction, RuntimeContext}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object Demo01ValueState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //创建配置文件对象
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    //    properties.setProperty("group.id", "csw2")
    //    properties.setProperty("auto.offset.reset", "earliest") //earliest：读取所有数据 latest：读取最新数据
    /**
      * 创建kafka消费者
      */
    val consumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("test_topicl", new SimpleStringSchema(), properties)

    consumer.setStartFromLatest()

    val kafkaDS: DataStream[String] = env.addSource(consumer)

    val kvDS: KeyedStream[(String, Int), String] = kafkaDS.map((_, 1)).keyBy(_._1)

    //    val countDS: DataStream[(String, Int)] = kvDS.sum(1)

    //使用自定义状态统计单词的数量
    val countDS: DataStream[(String, Int)] = kvDS.flatMap(new MyFlatMapFunction)

    countDS.print()

    env.execute()
  }
}

class MyFlatMapFunction extends RichFlatMapFunction[(String, Int), (String, Int)] {

  var valueState: ValueState[Int] = _

  /**
    *
    * 在flatMap之前执行
    * 在open 方法中定义状态
    *
    */
  override def open(parameters: Configuration): Unit = {
    //获取flink执行环境
    val context: RuntimeContext = getRuntimeContext

    //创建状态描述对象
    val valueStateDesc = new ValueStateDescriptor[Int]("count", classOf[Int])

    /**
      * 为每一个key 保存一个单一的值
      * 用于保存单词的数量
      */
    //获取状态
    valueState = context.getState(valueStateDesc)
  }

  override def flatMap(value: (String, Int), out: Collector[(String, Int)]): Unit = {

    //更新单词的数量,使用当前数量加上之前状态的数量得到新的数量
    //获取之前状态的值  即之前单词的个数
    val lastCount: Int = valueState.value()

    //更新状态， 更新单词的数量
    valueState.update(lastCount + value._2)

    //获取最新的数量
    val count: Int = valueState.value()

    //将数据发送到下游
    out.collect((value._1, count))
  }
}
